package com.sailing.lianxi.demoservice.onecusmer;
import java.util.Collection;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.common.TopicPartition;


public class MyConsumerRebalancerListener implements ConsumerRebalanceListener  {

	String storagePrefix = "storage2";
	private OffsetManager offsetManager = new OffsetManager(storagePrefix);
	
	private Consumer<String, String> consumer;
	
	public MyConsumerRebalancerListener(Consumer<String, String> consumer) {
		this.consumer = consumer;
	}

	public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
		for (TopicPartition partition : partitions) {
			offsetManager.saveOffsetInExternalStore(partition.topic(), partition.partition(), consumer.position(partition));
		}
		
	}

	public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
		for (TopicPartition partition : partitions) {
			consumer.seek(partition, offsetManager.readOffsetFromExternalStore(partition.topic(), partition.partition()));
		}
	}
	

}
